package com.wsg.dep.queue;

import com.wsg.dep.queue.seq.Seq;

/**
 * 生产-消费协调门控
 */
public class ProducerConsumerBarrier {
    private final Seq producerSeq = new Seq();   // 生产者最新序列号
    private final Seq consumerSeq = new Seq();   // 消费者最低进度（用于背压控制）
    private final int bufferSize;

    public ProducerConsumerBarrier(int bufferSize) {
        this.bufferSize = bufferSize;
    }

    /**
     * 生产者申请写入位置（含背压控制）
     */
    public long claimProducerSequence() {
        long sequence;
        do {
            sequence = producerSeq.get() + 1;
            // 背压控制：防止覆盖未消费数据
            while (sequence - consumerSeq.get() >= bufferSize) {
                Thread.yield();
            }
        } while (sequence < 0); // 防御性校验
        return sequence;
    }

    /**
     * 消费者获取可读范围 [nextSequence, availableSequence]
     */
    public long waitForConsumer(long nextSequence) {
        while (nextSequence > producerSeq.get()) {
            Thread.yield();
        }
        return producerSeq.get();
    }

    /**
     * 消费者提交处理完成的序列号
     */
    public void commitConsumerSequence(long sequence) {
        consumerSeq.set(sequence);
    }

    /**
     * 消费者提交处理完成的序列号
     */
    public void commitProduceSequence(long sequence) {
        producerSeq.set(sequence);
    }
    // Getter
    public Seq getProducerSeq() { return producerSeq; }
    public Seq getConsumerSeq() { return consumerSeq; }
}